package com.mydemo.charging.order.mqtt;

import com.alibaba.fastjson2.JSON;
import com.mydemo.charging.common.constant.MqttDestinationConst;
import com.mydemo.charging.common.pojo.message.ChargingCheckResultMessage;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.data.redis.core.RedisTemplate;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;

/**
 * 处理emqx的主题消息的唯一消费者
 * 订单中所有也业务消费都由这个类实现 需要最终处理的消费逻辑有2个
 * 1. 接收设备反馈信息
 * TODO 2. 接收设备充电进度信息
 */
@Slf4j
public class MqttConsumer implements MqttCallbackExtended {

    private MqttClient mqttClient;
    private MqttContext mqttContext;
    public MqttConsumer(MqttClient mqttClient,MqttContext mqttContext) {
        this.mqttContext=mqttContext;
        this.mqttClient = mqttClient;
    }

    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        //使用连接对象监听主题 接收设备反馈
        try{
            log.debug("mqtt开始订阅主题...");
            mqttClient.subscribe(MqttDestinationConst.GUN_CHECK_RESULT_TOPIC,1);
            log.debug("mqtt已经订阅主题:{}",MqttDestinationConst.GUN_CHECK_RESULT_TOPIC);
        }catch (Exception e){
            log.error("主题订阅异常",e);
        }
    }
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        //先不考虑消费逻辑如何执行 先检查流程跑通
        log.debug("主题:{},消息:{}",topic,new String(message.getPayload(), StandardCharsets.UTF_8));
        //1.解析消息 获得反序列化的对象
        ChargingCheckResultMessage msg=null;
        String strMsg = new String(message.getPayload(), StandardCharsets.UTF_8);
        msg=JSON.parseObject(strMsg,ChargingCheckResultMessage.class);
        //2.mqttContext调用获取get orderService 调用消费
        //2.1抢锁 setnx key(orderNo) "" EX 3
        RedisTemplate redisTemplate = mqttContext.getRedisTemplate();
        String lockKey="charging:device:result:"+msg.getOrderNo();
        Boolean getKey = redisTemplate.boundValueOps(lockKey)
                .setIfAbsent("", 3, TimeUnit.SECONDS);
//        try{
//            if (getKey){
//                log.debug("获取锁成功,开始处理消息");
//                mqttContext.getOrderConsumerService().handleCheckResult(msg);
//            }else{
//                log.debug("获取锁失败,结束消息消费");
//            }
//        }catch (Exception e){
//            log.error("处理消息异常",e);
//        }finally{
//            redisTemplate.delete(lockKey);
//        }
    }
    @Override
    public void connectionLost(Throwable cause) {
        log.error("mqtt连接断开",cause);
    }

    /**
     * 充电桩项目可以使用deliverComplete 打印生产发送的结果 qos=1 | 2生效 0 不会调用
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        log.debug("消息发送成功,消息id:{}",token.getMessageId());
    }
}
